package org.spring.springboot.kafka.xml.producer;

import java.io.IOException;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;


public class KafkaProducerImp extends Thread {

    private String topic;
    private Properties prop;
    private List<String> messages;

    public KafkaProducerImp(String topic , Properties prop, List<String> messages){
        super();
        this.topic = topic;
        this.prop = prop;
        this.messages = messages;
    }

    @Override
    public void run() {

        //获取开始时间
        long startTime=System.currentTimeMillis();

        try {
            KafkaProducer<String,String> producer = new KafkaProducer<String,String>(prop);
 
            for (String message : messages) {
                ProducerRecord<String, String> pr = new ProducerRecord<String, String>(topic, message);
                producer.send(pr);
            }
            
            TimeUnit.SECONDS.sleep(10);
            producer.close();
        }catch (Exception e) {
            e.printStackTrace();
        }

        //获取结束时间
        long endTime=System.currentTimeMillis();
        System.out.println("程序运行时间： "+(endTime-startTime)/1000+"ms");
    }

    public static void main(String[] args) throws IOException {

        Properties prop = new Properties();
        prop.load(KafkaProducerImp.class.getResourceAsStream("kafka.properties"));
        List<String> messages = null;
        String topics = prop.getProperty("topic");
        System.out.println("topicName--:" + topics);
        new KafkaProducerImp(topics,prop,messages).start();
    
    }
}